package drivers

import (
	"app/pkg/queue"
	"context"
	"encoding/json"
	"fmt"
	"os"
	"path/filepath"
	"strconv"
	"sync"
	"time"
)

// DiskDriver 基于文件系统的队列驱动结构体
type DiskDriver struct {
	baseDir string
	mu      sync.Mutex
}

// NewDisk 创建一个新的文件系统队列驱动实例
func NewDisk(baseDir string) *DiskDriver {
	// 创建基础目录（如果不存在）
	if err := os.MkdirAll(baseDir, 0755); err != nil {
		panic(fmt.Sprintf("Failed to create base directory: %v", err))
	}
	return &DiskDriver{
		baseDir: baseDir,
	}
}

// Connect 连接操作，文件系统驱动无需连接，直接返回 nil
func (d *DiskDriver) Connect() error {
	return nil
}

// Close 关闭操作，文件系统驱动无需关闭操作，直接返回 nil
func (d *DiskDriver) Close() error {
	return nil
}

// Topic 创建主题目录
func (d *DiskDriver) Topic(topic string) error {
	topicDir := filepath.Join(d.baseDir, topic)
	return os.MkdirAll(topicDir, 0755)
}

// Publish 将消息发布到指定主题的队列文件中
func (d *DiskDriver) Publish(ctx context.Context, topic string, message []byte, delay int) error {
	d.mu.Lock()
	defer d.mu.Unlock()

	topicDir := filepath.Join(d.baseDir, topic)
	if err := os.MkdirAll(topicDir, 0755); err != nil {
		return err
	}

	// 如果有延迟，使用不同的文件名表示延迟队列
	fileName := filepath.Join(topicDir, fmt.Sprintf("%d.msg", time.Now().UnixNano()))
	if delay > 0 {
		fileName = filepath.Join(topicDir, fmt.Sprintf("%d_delayed.msg", time.Now().Add(time.Duration(delay)*time.Second).UnixNano()))
	}

	return os.WriteFile(fileName, message, 0644)
}

// Subscribe 消费队列消息
func (d *DiskDriver) Subscribe(ctx context.Context, topic string, opt queue.Options, callback func(ctx context.Context, channel <-chan queue.QueueMessgae)) error {
	topicDir := filepath.Join(d.baseDir, topic)
	if err := os.MkdirAll(topicDir, 0755); err != nil {
		return err
	}

	msgChan := make(chan queue.QueueMessgae, 1)
	defer close(msgChan)

	// 启动协程处理消息
	go callback(ctx, msgChan)

	ticker := time.NewTicker(1 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return nil
		case <-ticker.C:
			// 处理延迟消息
			d.processDelayedMessages(ctx, topic)

			// 获取队列文件列表
			files, err := os.ReadDir(topicDir)
			if err != nil {
				continue
			}

			for _, file := range files {
				if file.IsDir() || filepath.Ext(file.Name()) != ".msg" {
					continue
				}

				filePath := filepath.Join(topicDir, file.Name())
				content, err := os.ReadFile(filePath)
				if err != nil {
					continue
				}

				var msg queue.QueueMessgae
				if err := json.Unmarshal(content, &msg); err != nil {
					continue
				}

				msgChan <- msg
			}
		}
	}
}

// processDelayedMessages 处理延迟消息
func (d *DiskDriver) processDelayedMessages(ctx context.Context, topic string) {
	d.mu.Lock()
	defer d.mu.Unlock()

	topicDir := filepath.Join(d.baseDir, topic)
	files, err := os.ReadDir(topicDir)
	if err != nil {
		return
	}

	now := time.Now().UnixNano()
	for _, file := range files {
		if file.IsDir() || !isDelayedMessage(file.Name()) {
			continue
		}

		timestamp, err := getTimestampFromDelayedMessage(file.Name())
		if err != nil {
			continue
		}

		if timestamp <= now {
			oldPath := filepath.Join(topicDir, file.Name())
			newPath := filepath.Join(topicDir, fmt.Sprintf("%d.msg", timestamp))
			if err := os.Rename(oldPath, newPath); err != nil {
				continue
			}
		}
	}
}

// Ack 确认消息，删除已处理的消息文件
func (d *DiskDriver) Ack(ctx context.Context, q queue.IQueue, qm queue.QueueMessgae) {
	d.mu.Lock()
	defer d.mu.Unlock()

	topicDir := filepath.Join(d.baseDir, q.Topic())
	fileName := findMessageFile(topicDir, qm)
	if fileName != "" {
		os.Remove(filepath.Join(topicDir, fileName))
	}
}

// Nack 未确认消息，增加投递次数并重新放入队列
func (d *DiskDriver) Nack(ctx context.Context, q queue.IQueue, qm queue.QueueMessgae) {
	d.mu.Lock()
	defer d.mu.Unlock()

	delivery := qm.GetHeader(queue.HeaderDelivery).(int)
	qm.SetHeader(queue.HeaderDelivery, delivery+1)

	topicDir := filepath.Join(d.baseDir, q.Topic())
	fileName := findMessageFile(topicDir, qm)
	if fileName != "" {
		os.Remove(filepath.Join(topicDir, fileName))
	}

	messageBytes, err := json.Marshal(qm)
	if err != nil {
		return
	}

	// 重新放入队列，可设置延迟时间
	if err := d.Publish(ctx, q.Topic(), messageBytes, 5); err != nil {
		return
	}
}

// Reject 拒绝消息，将消息放入死信队列
func (d *DiskDriver) Reject(ctx context.Context, q queue.IQueue, qm queue.QueueMessgae) {
	d.mu.Lock()
	defer d.mu.Unlock()

	if q.Dlq() {
		dlqDir := filepath.Join(d.baseDir, q.Topic(), "dlq")
		if err := os.MkdirAll(dlqDir, 0755); err != nil {
			return
		}

		messageBytes, err := json.Marshal(qm)
		if err != nil {
			return
		}

		fileName := filepath.Join(dlqDir, fmt.Sprintf("%d.msg", time.Now().UnixNano()))
		if err := os.WriteFile(fileName, messageBytes, 0644); err != nil {
			return
		}
	}

	// 从正常队列中删除消息
	d.Ack(ctx, q, qm)
}

// isDelayedMessage 判断是否为延迟消息文件
func isDelayedMessage(fileName string) bool {
	return filepath.Ext(fileName) == ".msg" && filepath.Base(fileName)[len(filepath.Base(fileName))-12:] == "_delayed.msg"
}

// getTimestampFromDelayedMessage 从延迟消息文件名中获取时间戳
func getTimestampFromDelayedMessage(fileName string) (int64, error) {
	timestampStr := fileName[:len(fileName)-12]
	return strconv.ParseInt(timestampStr, 10, 64)
}

// findMessageFile 根据消息查找对应的文件
func findMessageFile(topicDir string, qm queue.QueueMessgae) string {
	files, err := os.ReadDir(topicDir)
	if err != nil {
		return ""
	}

	for _, file := range files {
		if file.IsDir() || filepath.Ext(file.Name()) != ".msg" {
			continue
		}

		filePath := filepath.Join(topicDir, file.Name())
		content, err := os.ReadFile(filePath)
		if err != nil {
			continue
		}

		var msg queue.QueueMessgae
		if err := json.Unmarshal(content, &msg); err != nil {
			continue
		}

		if msg.GetId() == qm.GetId() {
			return file.Name()
		}
	}

	return ""
}
